package com.sharp;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.TupleImpl;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.Map;

/**
 * Hello world!
 *
 */
public class LocalMyTopology
{

    public static class MySpout extends BaseRichSpout{

        private Map map;
        private TopologyContext topologyContext;
        private SpoutOutputCollector spoutOutputCollector;
        @Override
        public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
            this.map = map;
            this.topologyContext = topologyContext;
            this.spoutOutputCollector = spoutOutputCollector;
        }

        int num = 0;
        @Override
        public void nextTuple() {
            this.spoutOutputCollector.emit(new Values(num++, "sharp"+num));
            Utils.sleep(2000);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("num","name"));
        }

        @Override
        public void ack(Object msgId) {
            super.ack(msgId);
            System.out.println("spout ack~~~~~~~~~~");
        }

        @Override
        public void fail(Object msgId) {
            super.fail(msgId);
            System.out.println("spout fail~~~~~~~~~~");
        }
    }

    public static class MyBlot extends BaseRichBolt{
        private Map stormConf;
        private TopologyContext context;
        private OutputCollector collector;
        @Override
        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
            this.stormConf = stormConf;
            this.context = context;
            this.collector = collector;
        }
        static Integer count = 0;
        @Override
        public void execute(Tuple input) {
            Integer num = input.getIntegerByField("num");
            String name = input.getStringByField("name");
            count += num;
            /*if(num%5 == 0){
                throw new RuntimeException("error");
            }*/
            System.out.println(name+"----->"+num);
            this.collector.ack(input);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {

        }

        @Override
        public Map<String, Object> getComponentConfiguration() {
            return super.getComponentConfiguration();
        }
    }
    public static void main( String[] args )
    {
        TopologyBuilder topology = new TopologyBuilder();
        topology.setSpout("spout_id", new MySpout(),1);
        topology.setBolt("bolt_id", new MyBlot(), 2).shuffleGrouping("spout_id");
        topology.setBolt("bolt_id2", new MyBlot(), 2).shuffleGrouping("spout_id");
        StormTopology topo = topology.createTopology();

        LocalCluster cluster = new LocalCluster();
        Config config = new Config();
        config.setNumWorkers(2);
        cluster.submitTopology(LocalMyTopology.class.getSimpleName(), config, topo);
    }
}
